
/*
 * Hlpay-Plus aggregate payment system.
 * Copyright (c) 2024-2025 Hlpay Team Copyright has the right of final interpretation.
 */

package com.hlkj.pay.listener;

import java.util.ArrayList;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.hlkj.pay.common.constants.MqConstant;
import com.hlkj.pay.dto.ad.AdStatisticsDto;
import com.hlkj.pay.dto.common.BusinessLogDto;
import com.hlkj.pay.infrastructure.model.admin.AdminUserBehaviorLogDO;
import com.hlkj.pay.infrastructure.model.common.RequestLogDO;
import com.hlkj.pay.infrastructure.model.pay.PayChannelRequestDO;
import com.hlkj.pay.infrastructure.model.user.UserRequestLogDO;
import com.hlkj.pay.service.common.IBusinessLogService;
import com.hlkj.pay.service.pay.IPayChannelService;

import lombok.extern.slf4j.Slf4j;

/**
 * @author monk@techsun.com
 * @date 2023/03/02 9:49
 */
@Slf4j
@Component
@ConditionalOnProperty(prefix = "rocketmq", name = "log_enabled", havingValue = "true")
public class LogBatchConsumer {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Autowired
    private IBusinessLogService businessLogService;

    @Autowired
    private IPayChannelService payChannelService;

    @Value("${rocketmq.consumer.pull-log-batch-size:100}")
    private Integer pullBatchSize;

    @Value("${rocketmq.consumer.consume-log-message-batch-max-size:100}")
    private Integer consumeMessageBatchMaxSize;

    @Value("${rocketmq.consumer.pull_interval:1000}")
    private Integer pullInterval;

    @Value("${server.port}")
    private String serverPort;

    @Bean
    public DefaultMQPushConsumer clientRequestMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.CLIENT_REQUEST_LOG_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.CLIENT_REQUEST_LOG_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.CLIENT_REQUEST_LOG_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() == 0) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("client localUserRequests receive messages: {}", msgs.size());
            List<UserRequestLogDO> localUserRequests = new ArrayList<>(msgs.size());
            log.debug("client localUserRequests receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                localUserRequests.add(JSONObject.parseObject(msg.getBody(), AdStatisticsDto.class));
            });
            log.info("localUserRequests size: {}", localUserRequests.size());
            // 处理批量消息
            businessLogService.consumeClientBusinessPramsLog(localUserRequests);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer behaviorMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.BEHAVIOR_LOG_RECORD_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.BEHAVIOR_LOG_RECORD_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.BEHAVIOR_LOG_RECORD_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() == 0) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("behavior requestLogList receive messages: {}", msgs.size());
            List<BusinessLogDto> requestLogList = new ArrayList<>(msgs.size());
            log.debug("behavior requestLogList receive receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                requestLogList.add(JSONObject.parseObject(msg.getBody(), BusinessLogDto.class));
            });
            log.info("behavior size: {}", requestLogList.size());
            // 处理批量消息
            businessLogService.consumeBusinessLog(requestLogList);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer behaviorParamsLogMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.BEHAVIOR_LOG_PARAMS_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.BEHAVIOR_LOG_PARAMS_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.BEHAVIOR_LOG_PARAMS_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() == 0) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("behavior params requestLogList receive messages: {}", msgs.size());
            List<AdminUserBehaviorLogDO> requestLogList = new ArrayList<>(msgs.size());
            log.debug("behavior params requestLogList receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                requestLogList.add(JSONObject.parseObject(msg.getBody(), AdminUserBehaviorLogDO.class));
            });
            log.info("behavior params size: {}", requestLogList.size());
            // 处理批量消息
            try {
                businessLogService.consumeBusinessPramsLog(requestLogList);
            }
            catch (Exception e) {
                e.printStackTrace();
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer merchantRequestMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.REQUEST_LOG_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.REQUEST_LOG_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.REQUEST_LOG_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() == 0) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("openapi requestLogList receive messages: {}", msgs.size());
            List<RequestLogDO> requestLogList = new ArrayList<>(msgs.size());
            log.debug("openapi requestLogList receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                requestLogList.add(JSONObject.parseObject(msg.getBody(), RequestLogDO.class));
            });
            log.info("requestLogList size: {}", requestLogList.size());
            // 处理批量消息
            businessLogService.consumeRequestLog(requestLogList);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer merchantLogMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConstant.MERCHANT_REQUEST_LOG_TOPIC);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe(MqConstant.MERCHANT_REQUEST_LOG_TOPIC, "*");
        // 设置每次消息拉取的时间间隔，单位毫秒
        consumer.setPullInterval(pullInterval);
        // 设置每个队列每次拉取的最大消息数
        consumer.setPullBatchSize(pullBatchSize);
        consumer.setInstanceName(MqConstant.MERCHANT_REQUEST_LOG_TOPIC + "_" + serverPort);
        // 设置消费者单次批量消费的消息数目上限
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (msgs.size() == 0) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            log.info("merchant request requestLogList receive messages: {}", msgs.size());
            List<PayChannelRequestDO> requestLogList = new ArrayList<>(msgs.size());
            log.debug("merchant request requestLogList receive topic: {}", msgs.get(0).getTopic());
            msgs.forEach(msg -> {
                requestLogList.add(JSONObject.parseObject(msg.getBody(), PayChannelRequestDO.class));
            });
            log.info("merchant request size: {}", requestLogList.size());
            // 处理批量消息
            payChannelService.consumerRequest(requestLogList);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }
}
